package org.raymond.icloud.modules.admin.consumer;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.raymond.icloud.apis.admin.entity.TSysLog;
import org.raymond.icloud.common.core.bean.LogBean;
import org.raymond.icloud.common.core.constants.MQConstants;
import org.raymond.icloud.modules.admin.service.ITSysLogService;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

@Slf4j
@Component
public class LogConsumer {
    @Autowired
    private ITSysLogService itSysLogService;

    @RabbitListener(id="listener-log", queues={MQConstants.QUEUEE_LOG_COMMON})
    @SendTo(MQConstants.EXCHANGE_LOG_COMMON+"/"+MQConstants.ROUTING_LOG)
    public void onMessage(List<Message> messageList, Channel channel) throws Exception {
        log.debug("log messageList:{}, channel:{}", messageList, channel);
        final List<Throwable> errorList = new ArrayList<>();
        List<TSysLog> logs = messageList.stream().map(msg->{
            MessageProperties messageProperties = msg.getMessageProperties();
            String encoding = messageProperties.getContentEncoding();
            if(StringUtils.isBlank(encoding)){
                encoding = "UTF-8";
            }
            String body = null;
            try {
                body = new String(msg.getBody(), encoding);
            } catch (UnsupportedEncodingException e) {
                errorList.add(e);
                return null;
            }
            LogBean logBean = JSONObject.parseObject(body, LogBean.class);
            TSysLog sysLog = new TSysLog();
            BeanUtils.copyProperties(logBean, sysLog);
            return sysLog;
        }).collect(Collectors.toList());

        if(errorList.size()>0) {
            log.error("encoding error count: {}, msg: {}", errorList.size(), errorList.get(0).getMessage());
            try {
                for(Message msg: messageList){
                    long deliveryTag = msg.getMessageProperties().getDeliveryTag();
                    channel.basicReject(deliveryTag, false);
                }
            }catch (Exception e){
                log.error(e.getMessage());
            }
            return;
        }

        logs = logs.stream().filter(Objects::nonNull).collect(Collectors.toList());
        itSysLogService.saveBatch(logs, logs.size());

        try {
            for(Message msg: messageList){
                long deliveryTag = msg.getMessageProperties().getDeliveryTag();
                channel.basicAck(deliveryTag, false);
            }
        }catch (Exception e){
            log.error(e.getMessage());
        }
    }

    @RabbitListener(id="listener-log-dlx", queues={MQConstants.QUEUEE_LOG_DLX})
    @SendTo(MQConstants.EXCHANGE_LOG_DLX+"/"+MQConstants.ROUTING_LOG_DLX)
    public void onDLX(Message message, Channel channel) throws IOException{
        MessageProperties messageProperties = message.getMessageProperties();
        long deliveryTag = messageProperties.getDeliveryTag();
        String encoding = messageProperties.getContentEncoding();
        if(StringUtils.isBlank(encoding)){
            encoding = "UTF-8";
        }
        String body = new String(message.getBody(), encoding);
        LogBean lb = JSONObject.parseObject(body, LogBean.class);
        if(lb!=null){
            log.info("ack");
            channel.basicAck(deliveryTag, false);
        }else{
            log.warn("reject");
            channel.basicReject(deliveryTag, false);
        }
    }
}
